package com.atguigu.flink.window;

import com.atguigu.flink.function.MyUtil;
import com.atguigu.flink.function.WaterSensorMapFunction;
import com.atguigu.flink.pojo.WaterSensor;
import org.apache.flink.api.common.functions.ReduceFunction;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.windowing.WindowFunction;
import org.apache.flink.streaming.api.windowing.assigners.TumblingProcessingTimeWindows;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
import org.apache.flink.util.Collector;

/**
 * Created by Smexy on 2023/2/27
 *
 *      基于时间的窗口，如何在滚动聚合中获取时间属性?
 *              全量聚合: process() TimeWindow timeWindow = context.window();
 *              滚动聚合:  reduce:
 *                          reduce(
 *                               ReduceFunction<T> reduceFunction,
 *                                  W: 窗口，基于时间的，W 就是 TimeWindow。 基于个数的，W 就是GlobalWindow
 *                               WindowFunction<T, R, K, W> function
 *                               )
 *
 *                      aggregate：
 *                               aggregate(
 *                                   AggregateFunction<T, ACC, V> aggFunction,
 *                                   WindowFunction<V, R, K, W> windowFunction
 *                                   )
 */
public class Demo8_GetTime
{
    public static void main(String[] args) {

        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

        //求每种传感器的水位和
        env
            .socketTextStream("hadoop103", 8888)
            .map(new WaterSensorMapFunction())
            .keyBy(WaterSensor::getId)
            .window(TumblingProcessingTimeWindows.of(Time.seconds(5)))
            //等窗口关闭后，才会把数据发往下游的print
            .reduce(
                new ReduceFunction<WaterSensor>()
                {
                    @Override
                    public WaterSensor reduce(WaterSensor value1, WaterSensor value2) throws Exception {

                        System.out.println("Demo6_Reduce.reduce");
                        value1.setVc(value1.getVc() + value2.getVc());
                        return value1;
                    }
                },
                /*
                            reduce( ReduceFunction<T> reduceFunction,
                                   W: 窗口，基于时间的，W 就是 TimeWindow。 基于个数的，W 就是GlobalWindow
                                WindowFunction<T, R, K, W> function
 *                              )
                    当前这个方法代表，每来一个元素，调用一次ReduceFunction。
                                    窗口关闭时，把ReduceFunction最后一次聚合的结果，作为输入传递给WindowFunction。
                    <T, R, K, W>
                        T 输入类型:
                        R 输出的类型
                        K key的类型
                        W 窗口类型

                        WindowFunction： 在窗口关闭时，调用一次！
                                            apply()是全量聚合！
                 */
                new WindowFunction<WaterSensor, String, String, TimeWindow>()
                {
                    // Iterable<WaterSensor> input : 只有一个元素，是reduceFunction最终聚合的结果
                    @Override
                    public void apply(String key, TimeWindow window, Iterable<WaterSensor> input, Collector<String> out) throws Exception {

                        System.out.println("Demo8_GetTime.apply");
                        WaterSensor waterSensor = input.iterator().next();
                        out.collect(MyUtil.parseTimeWindow(window) + ":" + key + ":" + waterSensor.getVc());

                    }
                }
            )
            .print();


        try {
            env.execute();
        } catch (Exception e) {
            e.printStackTrace();
        }

    }
}
